package kvraft

import (
	"bytes"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"../labgob"
	"../labrpc"
	"../raft"
)

const debug = 0

//DPrintf :
func DPrintf(format string, a ...interface{}) (n int, err error) {
	if debug > 0 {
		log.Printf(format, a...)
	}
	return
}

//Op struct :
type Op struct {
	// Your definitions here.
	// Field names must start with capital letters,
	// otherwise RPC will break.
	ClientID  int64
	ReqNum    int
	Operation string
	Key       string
	Value     string
}

//Result :
type Result struct {
	Err   Err
	value string
}

//KVServer struct :
type KVServer struct {
	mu      sync.Mutex
	me      int
	rf      *raft.Raft
	applyCh chan raft.ApplyMsg
	dead    int32 // set by Kill()

	maxraftstate int // snapshot if log grows this big

	// Your definitions here.
	database   map[string]string
	reqHistory map[int64]int

	requestCh chan Result
}

func (kv *KVServer) queryReqHistory(clientid int64, curNum int) bool {
	maxNum, ok := kv.reqHistory[clientid]
	if !ok {
		kv.reqHistory[clientid] = -1
		return false
	}
	return maxNum >= curNum

}

//Get :
func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	// Your code here.
	kv.mu.Lock()
	ok := kv.queryReqHistory(args.ClientID, args.ReqNum)
	kv.mu.Unlock()
	command := Op{args.ClientID, args.ReqNum, "Get", args.Key, ""}
	if !ok {
		//command := Op{args.ClientID, args.ReqNum, "Get", args.Key, ""}
		term1, _, isleader := kv.rf.Start(command)
		if isleader {
			//kv.mu.Unlock()
			var result Result
			select {
			case result = <-kv.requestCh:
			case <-time.After(150 * time.Millisecond):
				//log.Printf("kv[%v]: get time out\n", kv.me)
				reply.Err = ErrWrongLeader
				return
			}
			//time.Sleep(50 * time.Millisecond)
			if term2, _ := kv.rf.GetState(); term2 != term1 {
				reply.Err = ErrWrongLeader
				return
			}
			reply.Value = result.value
			if reply.Value == "" {
				reply.Err = ErrNoKey
			} else {
				reply.Err = OK
			}
		} else {
			reply.Err = ErrWrongLeader
		}
	} else {
		result := kv.database[args.Key]
		if result == "" {
			reply.Err = ErrNoKey
		} else {
			reply.Err = OK
		}
		reply.Value = result
	}
}

//PutAppend :
func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	// Your code here.
	kv.mu.Lock()
	ok := kv.queryReqHistory(args.ClientID, args.ReqNum)
	kv.mu.Unlock()
	command := Op{args.ClientID, args.ReqNum, args.Op, args.Key, args.Value}
	if !ok {
		term1, _, isleader := kv.rf.Start(command)
		if isleader {
			select {
			case <-kv.requestCh:
			case <-time.After(150 * time.Millisecond):
				//log.Printf("kv[%v]: putappend time out", kv.me)
				reply.Err = ErrWrongLeader
				return
			}
			if term2, _ := kv.rf.GetState(); term2 != term1 {
				reply.Err = ErrWrongLeader
				return
			}
			reply.Err = OK
		} else {
			reply.Err = ErrWrongLeader
		}
	} else {
		reply.Err = OK
	}

}

//applyRoutine :
func (kv *KVServer) applyRoutine() {
	for {
		applymsg := <-kv.applyCh
		if applymsg.CommandValid {
			kv.mu.Lock()
			//var op Op
			//DPrintf("[%v]: command %v", kv.me, applymsg.Command)
			if applymsg.CommandIndex == 0 {
				kv.mu.Unlock()
				continue
			}
			op, isOp := applymsg.Command.(Op)
			//if _, ok := kv.stateMachine[op.ID]; ok || !isOp { //duplicate
			if ok := kv.queryReqHistory(op.ClientID, op.ReqNum); ok || !isOp {
				kv.mu.Unlock()
				continue
			}
			DPrintf("[%v]: command %v", kv.me, applymsg.Command)
			var result Result
			switch op.Operation {
			case "Put":
				kv.database[op.Key] = op.Value
				result.Err = OK
				//DPrintf("[%v]: Put finish", kv.me)
			case "Append":
				tmp := kv.database[op.Key]
				kv.database[op.Key] = tmp + op.Value
				result.Err = OK
				//DPrintf("[%v]: Append finish", kv.me)
			case "Get":
				//DPrintf("[%v]: Get finish", kv.me)
				value, ok := kv.database[op.Key]
				if ok {
					result.value = value
				} else {
					result.value = ""
				}
			default:
			}
			kv.reqHistory[op.ClientID] = op.ReqNum
			kv.mu.Unlock()
			select {
			case kv.requestCh <- result: //try to send channel to putappend or get RPC
			default:
			}
		} else if applymsg.IsSnapshot {
			kv.readSnapshot(applymsg.Snapshots)
			DPrintf("kv[%v]: recv snapshot database %v\n", kv.me, kv.database)
		} else {
			snapshotmsg := raft.ApplyMsg{}
			if kv.maxraftstate != -1 && applymsg.RaftStateSize > kv.maxraftstate { //send snapshot
				DPrintf("[%v]: before createsnapshot", kv.me)
				writesnapshot := new(bytes.Buffer)
				enc := labgob.NewEncoder(writesnapshot)
				enc.Encode(kv.database)
				enc.Encode(kv.reqHistory)
				snapshot := writesnapshot.Bytes()
				snapshotmsg.CommandValid = true
				snapshotmsg.Snapshots = snapshot
				DPrintf("[%v]: after createsnapshot", kv.me)
			} else {
				snapshotmsg.CommandValid = false
			}
			kv.applyCh <- snapshotmsg //communicate with raft to confirm whether create snapshot or not and snapshot index
		}
	}
}

func (kv *KVServer) readSnapshot(data []byte) {
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var db map[string]string
	var reqhistory map[int64]int
	if d.Decode(&db) != nil ||
		d.Decode(&reqhistory) != nil {
		DPrintf("kv[%v]: readSnapshot error!", kv.me)
	} else {
		kv.database = db
		kv.reqHistory = reqhistory
	}
}

//Kill :
// the tester calls Kill() when a KVServer instance won't
// be needed again. for your convenience, we supply
// code to set rf.dead (without needing a lock),
// and a killed() method to test rf.dead in
// long-running loops. you can also add your own
// code to Kill(). you're not required to do anything
// about this, but it may be convenient (for example)
// to suppress debug output from a Kill()ed instance.
//
func (kv *KVServer) Kill() {
	atomic.StoreInt32(&kv.dead, 1)
	kv.rf.Kill()
	// Your code here, if desired.
}

func (kv *KVServer) killed() bool {
	z := atomic.LoadInt32(&kv.dead)
	return z == 1
}

//StartKVServer :
// servers[] contains the ports of the set of
// servers that will cooperate via Raft to
// form the fault-tolerant key/value service.
// me is the index of the current server in servers[].
// the k/v server should store snapshots through the underlying Raft
// implementation, which should call persister.SaveStateAndSnapshot() to
// atomically save the Raft state along with the snapshot.
// the k/v server should snapshot when Raft's saved state exceeds maxraftstate bytes,
// in order to allow Raft to garbage-collect its log. if maxraftstate is -1,
// you don't need to snapshot.
// StartKVServer() must return quickly, so it should start goroutines
// for any long-running work.
//
func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {
	// call labgob.Register on structures you want
	// Go's RPC library to marshall/unmarshall.
	labgob.Register(Op{})

	kv := new(KVServer)
	kv.me = me
	kv.maxraftstate = maxraftstate

	// You may need initialization code here.

	kv.applyCh = make(chan raft.ApplyMsg)
	kv.rf = raft.Make(servers, me, persister, kv.applyCh)

	// You may need initialization code here.
	kv.database = make(map[string]string)
	kv.reqHistory = make(map[int64]int)
	kv.requestCh = make(chan Result)
	kv.readSnapshot(persister.ReadSnapshot())
	go kv.applyRoutine()

	return kv
}
